package cn.me.alphamq.broker.core.timewheel;

import cn.me.alphamq.broker.cache.Container;
import cn.me.alphamq.broker.constant.BrokerConstant;
import cn.me.alphamq.broker.entity.RetryConsumeMsg;
import cn.me.alphamq.broker.enumeration.SlotMsgTypeEnum;
import cn.me.alphamq.broker.model.timewheel.DelayMessageModel;
import cn.me.alphamq.broker.model.timewheel.DelayTxMessageModel;
import cn.me.alphamq.broker.model.timewheel.TimeWheelSlotModel;
import cn.me.alphamq.broker.model.tx.TxMsgModel;
import cn.me.alphamq.broker.store.SlaveCtxManager;
import cn.me.alphamq.broker.store.TimeWheelModelManager;
import cn.me.alphamq.broker.store.TxMsgManager;
import cn.me.alphamq.common.entity.Message;
import cn.me.alphamq.common.entity.req.PushDelayMsgReq;
import cn.me.alphamq.common.entity.req.PushMsgReq;
import cn.me.alphamq.common.entity.req.TxCallbackCheckReq;
import cn.me.alphamq.common.enumeration.EventCodeEnum;
import cn.me.alphamq.common.msg.TcpMsg;
import cn.me.alphamq.common.util.TopicNameUtil;
import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 时间轮中到期消息的处理器
 *
 * @author f
 */
public class TimeWheelModelExecutor {

    private static final Logger log = LoggerFactory.getLogger(TimeWheelModelExecutor.class);

    /**
     * 处理到期的消息
     *
     * @param slot                      到期消息所在时间轮的槽位
     * @param shouldAckToCurrDelayQueue 这个参数仅用于第二个 if 语句，表示 是否应该向当前延迟队列写入 ack
     */
    public static void execute(TimeWheelSlotModel slot, boolean shouldAckToCurrDelayQueue) {
        int msgType = slot.getMsgType();
        if (SlotMsgTypeEnum.RETRY_MSG.getCode() == msgType) {
            RetryConsumeMsg retryConsumeMsg = (RetryConsumeMsg) slot.getMsg();
            String topic = retryConsumeMsg.getTopic();
            int queueId = retryConsumeMsg.getQueueId();
            String consumerGroup = retryConsumeMsg.getConsumerGroup();
            int offset = retryConsumeMsg.getMsgOffset();
            int length = retryConsumeMsg.getMsgLength();

            // 获取重试 topic，并且不能重复叠加重试 topic 的前缀
            String topicName = !TopicNameUtil.isRetryTopicName(topic) ?
                    TopicNameUtil.getRetryTopicName(consumerGroup) :
                    topic;
            try {
                // 读取 commitLog 的主体，并将 commitLog 重新写入重试消息的文件
                Container.getCommitLogAppendHandler().appendMsg(Message.newRetryMsg(
                        topicName, queueId,
                        Container.getCommitLogMMapFileModelManager().get(topic).readContent(offset, length),
                        retryConsumeMsg.getCurRetryTimes()
                ));
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        } else if (SlotMsgTypeEnum.DELAY_MSG.getCode() == msgType) {
            // 将消息写入 commitLog 文件
            PushDelayMsgReq pushDelayMsgReq = (PushDelayMsgReq) slot.getMsg();
            try {
                Container.getCommitLogAppendHandler().appendMsg(new Message(pushDelayMsgReq));
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }

            // 只有主节点才可能收到 推送延迟消息 的请求，发送同步请求给从节点
            SlaveCtxManager.writeAndFlush(EventCodeEnum.PUSH_MSG_REQ, new PushMsgReq(pushDelayMsgReq));

            // 通过 ack 表示这条消息已经完成处理
            if (shouldAckToCurrDelayQueue) {
                Container.getConsumeQueueConsumeHandler().ack(BrokerConstant.DELAY_TOPIC_NAME,
                        BrokerConstant.RECOVER_DELAY_MSG_CONSUMER_GROUP,
                        Container.getBrokerProperties().getBrokerCurrDelayQueueId()
                );
            }
        } else if (SlotMsgTypeEnum.TX_MSG.getCode() == msgType) {
            DelayTxMessageModel delayTxMessageModel = (DelayTxMessageModel) slot.getMsg();
            String msgId = delayTxMessageModel.getMsgId();

            // 如果事务消息模型为 null，则说明事务消息已被处理，跳过即可
            TxMsgModel txMsgModel = TxMsgManager.get(msgId);
            if (txMsgModel == null) {
                return;
            }

            // 如果与发送事务消息的客户端断开连接，则跳过它
            ChannelHandlerContext ctx = txMsgModel.getCtx();
            if (!ctx.channel().isActive()) {
                log.error("producer [{}] has disconnected!", msgId);
                TxMsgManager.remove(msgId);
                return;
            }

            // 给生产者发送回调检查消息
            ctx.writeAndFlush(new TcpMsg(EventCodeEnum.TX_CALLBACK_CHECK_REQ, JSON.toJSONBytes(
                    new TxCallbackCheckReq(msgId, txMsgModel.getTopic(), txMsgModel.getQueueId(), txMsgModel.getBody(),
                            txMsgModel.getTxFlag(), txMsgModel.getTransactionState(), txMsgModel.getProducerId()
                    ))));

            // 将回调检查的延迟消息放入延迟队列中 TODO 设置最大延迟次数，就和重试消息一样
            TimeWheelModelManager.add(DelayMessageModel.newDelayTxMsgModel(msgId));
        }
    }
}
